package com.zg.CliemtManage;

import com.zg.domain.DataForClient;
import com.zg.domain.LotClient;
import com.zg.utils.ThreadPool;
import com.zg.webSocket.SendAllMsg;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;


import java.util.HashMap;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;


@Slf4j
public class ClientInfo {

    //线程池的获取
    private ExecutorService threadPool = ThreadPool.getThreadPool();

    //处理主题相关的方法
    private ClientTopics clientTopics = new ClientTopics();

    //在线设备
    private ConcurrentHashMap<Channel,String> onlineMap = DataForClient.getOnlineMap();

    //在线设备详细信息
    private ConcurrentHashMap<String, LotClient> clientDetail = DataForClient.getClientDetail();

    /**
     * 设备上线处理
     * @param clientId
     * @param channel
     */
    public void client_accept(String clientId, Channel channel){
        LotClient lotClient = new LotClient();
        lotClient.setClient_id(clientId);
        lotClient.setChannel(channel);
        clientDetail.put(clientId,lotClient);
        onlineMap.put(channel,clientId);
        onlineMsgToWebSocket();//发送在线消息给websocket端
        log.info("添加客户端：{}，上线消息,共有：{}个在线设备",clientId,onlineMap.size());
    }

    /**
     * 设备下线处理
     * 或者发生异常退出
     * @param channel
     */
    public void client_shutdown(Channel channel){
        if (onlineMap.containsKey(channel)){
            String clientId = onlineMap.get(channel);
            LotClient lotClient = clientDetail.get(clientId);
            clientDetail.remove(clientId);//移除在线设备详细信息
            onlineMap.remove(channel);//移除在线设备
            clientTopics.delClientAllTopic(lotClient);//移除设备主题推送
            channel.close();
            onlineMsgToWebSocket();//发送在线消息给websocket端
        }
    }



    /**
     * 添加主题订阅
     * @param topic
     * @param channel
     */
    public void addClientTopic(Set<String> topic, Channel channel){
        //在详细信息中添加主题
        String s = onlineMap.get(channel);
        LotClient lotClient = clientDetail.get(s);
        List<String> topics = lotClient.getTopics();
        //添加主题
        for (String s1 : topic) {
            topics.add(s1);
        }
        clientTopics.addTopic(topic,channel);
    }

    /**
     * 客户端取消订阅主题
     * @param channel
     * @param topics
     */
    public void delClientTopic(Channel channel, List<String> topics){
        if (onlineMap.containsKey(channel)){
            String client_id = onlineMap.get(channel);
            LotClient lotClient = clientDetail.get(client_id);
            List<String> subTopic = lotClient.getTopics();
            for (String topic : topics) {
                subTopic.remove(topic);
            }
            clientTopics.delSomeTopic(channel,topics);
        }
    }

    /**
     * 转发消息到订阅了客户端
     * @param topic
     * @param data
     */
    public void relayMessage(String topic,Object data){
        threadPool.submit(new Runnable() {
            @Override
            public void run() {
                clientTopics.sendTopicMsg(topic,data);
            }
        });

    }



    /**
     * 管理员可以剔除非法客户端,强制下线
     * @param clientId 要移除的设备id
     */
    public void admin_delOnlineClient(String clientId){
        if (clientDetail.containsKey(clientId)){
            LotClient lotClient = clientDetail.get(clientId);
            this.client_shutdown(lotClient.getChannel());
        }
    }

    /**
     * 管理员移除设备订阅消息
     * @param clientId  要移除的设备id
     * @param topics  要移除的主题
     */
    public void admin_delSubTopic(String clientId,List<String> topics){
        if (clientDetail.containsKey(clientId)){
            LotClient lotClient = clientDetail.get(clientId);
            this.delClientTopic(lotClient.getChannel(),topics);
        }
    }

    /**
     * 将 设备的在线数量通过websocket实时的传输给前端
     */
    private void onlineMsgToWebSocket(){
        threadPool.submit(new Runnable() {
            @Override
            public void run() {
                HashMap<String, Object> map = new HashMap<>();
                map.put("onlineNum",onlineMap.size());
                map.put("time",System.currentTimeMillis());
                SendAllMsg.sendInfo(map);
            }
        });
    }

    /**
     * 管理员向指定的客户端发送消息
     */
    public void sendMsgToClient(String client_id,String topic,Object data){
        threadPool.submit(new Runnable() {
            @Override
            public void run() {
                if (clientDetail.containsKey(client_id)){
                    LotClient lotClient = clientDetail.get(client_id);
                    for (String lotClientTopic : lotClient.getTopics()) {
                        if (lotClientTopic.equals(topic)){
                            clientTopics.sendMsgToClient(topic,data,lotClient.getChannel());
                        }
                    }

                }

            }
        });
    }

}
